package drds.server.execute_engine.data_handling;

import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.memory.unsafe.utils.BytesTools;
import drds.server.net.packet.RowDataPacket;

public class RowDataPacketSorter {

	private static final Logger LOGGER = LoggerFactory.getLogger(RowDataPacketSorter.class);
	protected final OrderByColum[] orderByColums;

	private Collection<RowDataPacket> sortedRowDataPacketQueue = new ConcurrentLinkedQueue<RowDataPacket>();
	private RowDataPacket[] array, resultTemp;
	private int $fromStartIndex, $from0, $fromMiddleIndexAdd1;

	public RowDataPacketSorter(OrderByColum[] orderByColums) {
		super();
		this.orderByColums = orderByColums;
	}

	public boolean addRowDataPacket(RowDataPacket rowDataPacket) {
		return this.sortedRowDataPacketQueue.add(rowDataPacket);

	}

	public Collection<RowDataPacket> getSortedRowDataPacketList() {
		try {
			this.mergeSort(sortedRowDataPacketQueue.toArray(new RowDataPacket[sortedRowDataPacketQueue.size()]));
		} catch (Exception e) {
			LOGGER.error("getSortedResultError", e);
		}
		if (array != null) {
			Collections.addAll(this.sortedRowDataPacketQueue, array);
		}

		return sortedRowDataPacketQueue;
	}

	private RowDataPacket[] mergeSort(RowDataPacket[] rowDataPackets) throws Exception {
		this.sortedRowDataPacketQueue.clear();
		array = rowDataPackets;
		if (rowDataPackets == null || rowDataPackets.length < 2 || this.orderByColums == null || orderByColums.length < 1) {
			return rowDataPackets;
		}
		merge(0, rowDataPackets.length - 1);

		return array;
	}

	private void merge(int startIndex, int endIndex) {
		if (startIndex < endIndex) {
			int middle = (startIndex + endIndex) / 2;

			merge(startIndex, middle);

			merge(middle + 1, endIndex);

			merge(startIndex, middle, endIndex);
		}
	}

	private void merge(int startIndex, int middleIndex, int endIndex) {
		resultTemp = new RowDataPacket[(endIndex - startIndex + 1)];

		$from0 = 0;
		$fromStartIndex = startIndex;
		$fromMiddleIndexAdd1 = middleIndex + 1;
		while ($fromStartIndex <= middleIndex || $fromMiddleIndexAdd1 <= endIndex) {
			if ($fromStartIndex == middleIndex + 1) {
				while ($fromMiddleIndexAdd1 <= endIndex) {
					resultTemp[$from0++] = array[$fromMiddleIndexAdd1++];

				}
			} else if ($fromMiddleIndexAdd1 == endIndex + 1) {
				while ($fromStartIndex <= middleIndex) {
					resultTemp[$from0++] = array[$fromStartIndex++];
				}

			} else {
				compare(0);
			}
		}
		for ($fromStartIndex = startIndex, $fromMiddleIndexAdd1 = 0; $fromStartIndex <= endIndex; $fromStartIndex++, $fromMiddleIndexAdd1++) {
			array[$fromStartIndex] = resultTemp[$fromMiddleIndexAdd1];
		}
	}

	/**
	 * 递归按照排序字段进行排序
	 */
	private void compare(int orderByColumIndex) {

		if (orderByColumIndex == this.orderByColums.length) {
			if (this.orderByColums[orderByColumIndex - 1].orderType == OrderByColum.ORDER_TYPE_ASC) {

				resultTemp[$from0++] = array[$fromStartIndex++];
			} else {
				resultTemp[$from0++] = array[$fromMiddleIndexAdd1++];
			}
			return;
		}

		byte[] left = array[$fromStartIndex].fieldBytesList.get(this.orderByColums[orderByColumIndex].colum.columIndex);
		byte[] right = array[$fromMiddleIndexAdd1].fieldBytesList.get(this.orderByColums[orderByColumIndex].colum.columIndex);

		if (compareObject(left, right, this.orderByColums[orderByColumIndex]) <= 0) {
			if (compareObject(left, right, this.orderByColums[orderByColumIndex]) < 0) {
				if (this.orderByColums[orderByColumIndex].orderType == OrderByColum.ORDER_TYPE_ASC) {// 升序
					resultTemp[$from0++] = array[$fromStartIndex++];
				} else {
					resultTemp[$from0++] = array[$fromMiddleIndexAdd1++];
				}
			} else {// 如果当前字段相等，则按照下一个字段排序
				compare(orderByColumIndex + 1);

			}

		} else {
			if (this.orderByColums[orderByColumIndex].orderType == OrderByColum.ORDER_TYPE_ASC) {// 升序
				resultTemp[$from0++] = array[$fromMiddleIndexAdd1++];
			} else {
				resultTemp[$from0++] = array[$fromStartIndex++];
			}

		}
	}

	public static final int compareObject(Object left, Object right, OrderByColum orderCol) {
		return compareObject((byte[]) left, (byte[]) right, orderCol);
	}

	public static final int compareObject(byte[] left, byte[] right, OrderByColum orderCol) {
		int columType = orderCol.getColum().getColumType();
		switch (columType) {
		case Column.COL_TYPE_DECIMAL:
		case Column.COL_TYPE_INT:
		case Column.COL_TYPE_SHORT:
		case Column.COL_TYPE_LONG:
		case Column.COL_TYPE_FLOAT:
		case Column.COL_TYPE_DOUBLE:
		case Column.COL_TYPE_LONGLONG:
		case Column.COL_TYPE_INT24:
		case Column.COL_TYPE_NEWDECIMAL:
			// 因为mysql的日期也是数字字符串方式表达，因此可以跟整数等一起对待
		case Column.COL_TYPE_DATE:
		case Column.COL_TYPE_TIMSTAMP:
		case Column.COL_TYPE_TIME:
		case Column.COL_TYPE_YEAR:
		case Column.COL_TYPE_DATETIME:
		case Column.COL_TYPE_NEWDATE:
		case Column.COL_TYPE_BIT:
			return BytesTools.compareTo(left, right);
		case Column.COL_TYPE_VAR_STRING:
		case Column.COL_TYPE_STRING:
			// ENUM和SET类型都是字符串，按字符串处理
		case Column.COL_TYPE_ENUM:
		case Column.COL_TYPE_SET:
			return BytesTools.compareTo(left, right);
			// BLOB相关类型和GEOMETRY类型不支持排序，略掉
		}
		return 0;
	}
}